package com.bjbn.service;

import com.bjbn.model.VehicleData;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Service
@Slf4j
public class CsvReaderService {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @Value("${csv.directory}")
    private String csvDirectory;

    @Value("${csv.topic}")
    private String topic;

    @Value("${csv.vehicle.count:50}")
    private int vehicleCount;

    @Value("${csv.coordinate.offset:0.001}")
    private double coordinateOffset;

    // 存储原始CSV数据
    private List<VehicleData> originalData = new ArrayList<>();
    // 存储每个虚拟车辆的数据
    private Map<String, List<VehicleData>> vehicleDataMap = new ConcurrentHashMap<>();
    // 每个虚拟车辆的当前索引
    private Map<String, AtomicInteger> vehicleIndexMap = new ConcurrentHashMap<>();
    // 每个虚拟车辆的运行状态
    private Map<String, Boolean> vehicleRunningMap = new ConcurrentHashMap<>();
    // 总记录数
    private final AtomicLong totalRecords = new AtomicLong(0);
    // 线程池用于处理多个虚拟车辆
    private ExecutorService executorService;

    // 时间格式化器
    private final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // JSON解析器
    private final ObjectMapper objectMapper = new ObjectMapper();

    // 随机数生成器
    private final Random random = new Random();

    @PostConstruct
    public void init() {
        loadCsvData();
    }

    public void loadCsvData() {
        log.info("Loading CSV data from directory: {}", csvDirectory);

        try {
            Path directory = Paths.get(csvDirectory);
            if (!Files.exists(directory)) {
                log.error("CSV directory does not exist: {}", csvDirectory);
                return;
            }

            File[] csvFiles = directory.toFile().listFiles((dir, name) -> name.toLowerCase().endsWith(".csv"));

            if (csvFiles == null || csvFiles.length == 0) {
                log.warn("No CSV files found in directory: {}", csvDirectory);
                return;
            }

            log.info("Found {} CSV files to load", csvFiles.length);

            // 清空现有数据
            originalData.clear();
            vehicleDataMap.clear();
            vehicleIndexMap.clear();
            vehicleRunningMap.clear();
            totalRecords.set(0);

            // 加载第一个CSV文件作为基础数据
            loadCsvFile(csvFiles[0]);

            // 基于原始数据生成多个虚拟车辆
            generateVirtualVehicles();

            // 创建线程池，线程数等于虚拟车辆数量，确保每个车辆都有独立线程
            if (executorService != null) {
                executorService.shutdown();
            }
            executorService = Executors.newFixedThreadPool(vehicleCount);

            log.info("Generated {} virtual vehicles with {} total records, created {} threads",
                    vehicleDataMap.size(), totalRecords.get(), vehicleCount);

        } catch (Exception e) {
            log.error("Error loading CSV files: {}", e.getMessage(), e);
        }
    }

    private void loadCsvFile(File csvFile) {
        log.info("Loading CSV file: {}", csvFile.getName());

        try (FileReader reader = new FileReader(csvFile);
             CSVParser csvParser = new CSVParser(reader, CSVFormat.DEFAULT.withFirstRecordAsHeader())) {

            for (CSVRecord record : csvParser) {
                VehicleData data = parseCsvRecord(record);
                originalData.add(data);
            }

            log.info("Loaded {} original records from file {}", originalData.size(), csvFile.getName());

        } catch (IOException e) {
            log.error("Error loading CSV file {}: {}", csvFile.getName(), e.getMessage(), e);
        }
    }

    private void generateVirtualVehicles() {
        log.info("Generating {} virtual vehicles from original data", vehicleCount);

        for (int i = 1; i <= vehicleCount; i++) {
            String vehicleId = String.format("民航-%04d", i);
            List<VehicleData> vehicleData = new ArrayList<>();

            for (VehicleData originalData : originalData) {
                VehicleData clonedData = cloneAndModifyVehicleData(originalData, vehicleId, i);
                vehicleData.add(clonedData);
            }

            vehicleDataMap.put(vehicleId, vehicleData);
            vehicleIndexMap.put(vehicleId, new AtomicInteger(0));
            vehicleRunningMap.put(vehicleId, false);

            totalRecords.addAndGet(vehicleData.size());
        }

        log.info("Generated {} virtual vehicles, each with {} records",
                vehicleDataMap.size(), originalData.size());
    }

    private VehicleData cloneAndModifyVehicleData(VehicleData original, String newVehicleId, int vehicleIndex) {
        VehicleData cloned = new VehicleData();

        // 复制所有字段
        cloned.setVT_NAME(newVehicleId);
        cloned.setEMPLOYEEID(original.getEMPLOYEEID());
        cloned.setVH_CODE(newVehicleId);
        cloned.setSTATE(original.getSTATE());
        cloned.setGEOX(original.getGEOX());
        cloned.setGEOY(original.getGEOY());
        cloned.setGEOH(original.getGEOH());
        cloned.setDIRECTION(original.getDIRECTION());
        cloned.setLL_TIME(original.getLL_TIME());
        cloned.setSPEED(original.getSPEED());
        cloned.setVEHICLE_TYPE(original.getVEHICLE_TYPE());
        cloned.setVI_REMARK(original.getVI_REMARK());
        cloned.setONLINE(original.getONLINE());
        cloned.setGROUP_NAME(original.getGROUP_NAME());
        cloned.setSIMCODE(original.getSIMCODE());
        cloned.setSIGN(original.getSIGN());
        cloned.setLOCALCOORDY(original.getLOCALCOORDY());
        cloned.setLOCALCOORDX(original.getLOCALCOORDX());
        cloned.setPOSITIONSTATE(original.getPOSITIONSTATE());

        // 更新LABELPARAMS中的车辆编号
        Map<String, Object> updatedLabelParams = new HashMap<>();
        if (original.getLABELPARAMS() != null) {
            for (Map.Entry<String, Object> entry : original.getLABELPARAMS().entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                // 如果值是字符串且包含车辆编号，则替换为新的车辆编号
                if (value instanceof String) {
                    String strValue = (String) value;
                    // 替换车辆编号，支持多种格式
                    // String updatedValue = strValue.replaceAll("车\\d+", newVehicleId);
                    updatedLabelParams.put(key, newVehicleId);
                } else {
                    updatedLabelParams.put(key, value);
                }
            }
        }
        cloned.setLABELPARAMS(updatedLabelParams);

        // 为每个虚拟车辆添加坐标偏移，避免轨迹完全重叠
        try {
            double originalGEOX = Double.parseDouble(original.getGEOX());
            double originalGEOY = Double.parseDouble(original.getGEOY());

            // 根据车辆索引计算偏移量，确保每个车辆的轨迹都有轻微差异
            double offsetX = (vehicleIndex % 10) * coordinateOffset;
            double offsetY = (vehicleIndex / 10) * coordinateOffset;

            cloned.setGEOX(String.valueOf(originalGEOX + offsetX));
            cloned.setGEOY(String.valueOf(originalGEOY + offsetY));
        } catch (NumberFormatException e) {
            log.warn("Failed to parse coordinates for vehicle {}, using original values", newVehicleId);
        }

        return cloned;
    }

    private VehicleData parseCsvRecord(CSVRecord record) {
        return new VehicleData(
                getValue(record, "VT_NAME"),
                getValue(record, "EMPLOYEEID"),
                getValue(record, "VH_CODE"),
                getIntValue(record, "STATE"),
                getValue(record, "GEOX"),
                getValue(record, "GEOY"),
                getValue(record, "GEOH"),
                getValue(record, "DIRECTION"),
                getValue(record, "LL_TIME"),
                getValue(record, "SPEED"),
                getValue(record, "VEHICLE_TYPE"),
                getValue(record, "VI_REMARK"),
                getValue(record, "ONLINE"),
                getValue(record, "GROUP_NAME"),
                getValue(record, "SIMCODE"),
                getValue(record, "SIGN"),
                getValue(record, "LOCALCOORDY"),
                getValue(record, "LOCALCOORDX"),
                getIntValue(record, "POSITIONSTATE"),
                parseLabelParams(getValue(record, "LABELPARAMS")),
                getValue(record, "LABELPATHKEY"),
                getValue(record, "CATEGORY")
        );
    }

    private Map<String, Object> parseLabelParams(String labelParamsStr) {
        if (labelParamsStr == null || labelParamsStr.trim().isEmpty()) {
            return new HashMap<>();
        }

        try {
            // 处理Python字典格式的字符串，转换为JSON格式
            String jsonStr = convertPythonDictToJson(labelParamsStr);

            // 尝试解析JSON字符串为Map对象
            return objectMapper.readValue(jsonStr, new TypeReference<Map<String, Object>>() {
            });
        } catch (Exception e) {
            log.warn("Failed to parse LABELPARAMS as JSON: {}. Using as string.", labelParamsStr);
            // 如果解析失败，创建一个包含原始字符串的Map
            Map<String, Object> fallbackMap = new HashMap<>();
            fallbackMap.put("text", labelParamsStr);
            return fallbackMap;
        }
    }

    private String convertPythonDictToJson(String pythonDictStr) {
        if (pythonDictStr == null || pythonDictStr.trim().isEmpty()) {
            return "{}";
        }

        // 移除首尾的空白字符
        String trimmed = pythonDictStr.trim();

        // 如果已经是JSON格式，直接返回
        if (trimmed.startsWith("{") && trimmed.endsWith("}")) {
            // 检查是否包含单引号（Python格式）
            if (trimmed.contains("'")) {
                // 将单引号替换为双引号
                return trimmed.replace("'", "\"");
            }
            return trimmed;
        }

        // 如果不是字典格式，返回空对象
        return "{}";
    }

    private String getValue(CSVRecord record, String columnName) {
        try {
            return record.get(columnName);
        } catch (IllegalArgumentException e) {
            log.warn("Column '{}' not found in CSV record", columnName);
            return "";
        }
    }

    private Integer getIntValue(CSVRecord record, String columnName) {
        try {
            return Integer.parseInt(record.get(columnName));
        } catch (IllegalArgumentException e) {
            log.warn("Column '{}' not found in CSV record", columnName);
            return null;
        }
    }

    @Async
    public void startContinuousProcessing() {
        if (vehicleDataMap.isEmpty()) {
            log.error("No data loaded. Please load CSV files first.");
            return;
        }

        log.info("Starting continuous processing for {} virtual vehicles", vehicleDataMap.size());

        // 为每个虚拟车辆启动独立的处理线程
        for (String vehicleId : vehicleDataMap.keySet()) {
            executorService.submit(() -> processVehicleData(vehicleId));
        }
    }

    private void processVehicleData(String vehicleId) {
        List<VehicleData> vehicleData = vehicleDataMap.get(vehicleId);
        AtomicInteger currentIndex = vehicleIndexMap.get(vehicleId);

        if (vehicleData == null || vehicleData.isEmpty()) {
            log.warn("No data found for vehicle: {}", vehicleId);
            return;
        }

        // 设置车辆为运行状态
        vehicleRunningMap.put(vehicleId, true);

        // 随机延迟0-10分钟
        int randomDelayMinutes = random.nextInt(11); // 0-10分钟
        int randomDelaySeconds = random.nextInt(60); // 0-59秒
        long totalDelayMs = (randomDelayMinutes * 60 + randomDelaySeconds) * 1000L;

        log.info("Vehicle {} will start processing after {} minutes {} seconds delay",
                vehicleId, randomDelayMinutes, randomDelaySeconds);

        try {
            Thread.sleep(totalDelayMs);
        } catch (InterruptedException e) {
            log.info("Vehicle {} processing interrupted during delay", vehicleId);
            Thread.currentThread().interrupt();
            return;
        }

        log.info("Starting continuous processing for vehicle: {} with {} records", vehicleId, vehicleData.size());

        while (vehicleRunningMap.get(vehicleId)) {
            try {
                // 获取当前时间作为推送时间
                String currentTime = LocalDateTime.now().format(formatter);

                // 获取当前索引的数据
                int index = currentIndex.getAndIncrement() % vehicleData.size();
                VehicleData data = vehicleData.get(index);

                // 更新LL_TIME为当前推送时间
                data.setLL_TIME(currentTime);

                // 使用车辆ID作为消息key
                String messageKey = vehicleId;
                kafkaProducerService.sendMessage(messageKey, data);

                log.debug("Sent record {}/{} from vehicle {} to Kafka at {}",
                        index + 1, vehicleData.size(), vehicleId, currentTime);

                // 等待1秒
                Thread.sleep(1000);

            } catch (InterruptedException e) {
                log.info("Vehicle {} processing interrupted", vehicleId);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Error during processing vehicle {}: {}", vehicleId, e.getMessage(), e);
            }
        }

        log.info("Stopped continuous processing for vehicle: {}", vehicleId);
    }

    public void stopContinuousProcessing() {
        log.info("Stopping continuous processing for all vehicles");
        for (String vehicleId : vehicleRunningMap.keySet()) {
            vehicleRunningMap.put(vehicleId, false);
        }
    }

    public boolean isRunning() {
        return vehicleRunningMap.values().stream().anyMatch(running -> running);
    }

    public long getTotalRecords() {
        return totalRecords.get();
    }

    public Map<String, Object> getDetailedStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("isRunning", isRunning());
        status.put("totalRecords", totalRecords.get());
        status.put("totalVehicles", vehicleDataMap.size());
        status.put("originalRecords", originalData.size());

        Map<String, Object> vehicleStatus = new HashMap<>();
        for (String vehicleId : vehicleDataMap.keySet()) {
            Map<String, Object> vehicleInfo = new HashMap<>();
            vehicleInfo.put("isRunning", vehicleRunningMap.get(vehicleId));
            vehicleInfo.put("currentIndex", vehicleIndexMap.get(vehicleId).get());
            vehicleInfo.put("totalRecords", vehicleDataMap.get(vehicleId).size());
            vehicleStatus.put(vehicleId, vehicleInfo);
        }
        status.put("vehicleStatus", vehicleStatus);

        return status;
    }

    public void reloadData() {
        log.info("Reloading CSV data...");
        loadCsvData();
    }

    public Set<String> getLoadedFiles() {
        return new HashSet<>(vehicleDataMap.keySet());
    }

    @PreDestroy
    public void destroy() {
        if (executorService != null && !executorService.isShutdown()) {
            log.info("Shutting down executor service...");
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
} 